package com.shujia.spark

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2WordCountSubmit {

  def main(args: Array[String]): Unit = {

    /**
      *
      * 将spark代码提交到服务器中运行
      * 1、删除本地   conf.setMaster("local")
      * 2、修改输入输出路径
      *
      * 将项目打包上传到服务器
      * 提交任务
      * spark-submit --class  com.shujia.spark.Demo2WordCountSubmit --master yarn-client   spark-1.0.jar
      *
      */


    /**
      * 创建spark运行环境
      *
      */
    //配置文件对象
    val conf = new SparkConf()

    //会在yarn中显示的名称
    conf.setAppName("wc")

    //运行方式，local:本地运行，
    //    conf.setMaster("local")


    //构建spark 上下文对象
    val sc: SparkContext = new SparkContext(conf)


    //1、读取数据

    /**
      * RDD: 弹性的分布式数据集（可以看作是scala中的一个集合来使用）
      *
      */

    //将读取文件的路径修改为hdfs的路径
    val linesRDD: RDD[String] = sc.textFile("/data/words")


    //2、将每一行的多个单词拆分出来
    //在spark中数据处理的方法一般称为算子，算子处理完了之后会返回一个新的RDD
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))


    //3、按照单词进行分组，在底层会产生shuffle
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)


    //4、统计单词的数量
    val countRDD: RDD[String] = groupRDD.map(kv => {
      val word: String = kv._1

      //一个组内所有单词
      val iter: Iterable[String] = kv._2

      //单词的数量
      val count: Int = iter.size

      //返回
      word + "," + count

    })


    ///删除输出路径
    //使用hadoop的api
    val configuration = new Configuration()
    val fileSystem: FileSystem = FileSystem.get(configuration)

    if (fileSystem.exists(new Path("/data/count"))) {
      fileSystem.delete(new Path("/data/count"), true)
    }


    //5、保存数据
    countRDD.saveAsTextFile("/data/count")


  }

}
